package com.ds.infrastructure.hyperspace.container.internal;

import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.ds.infrastructure.hyperspace.common.EndpointProtocolEnum;
import com.ds.infrastructure.hyperspace.common.Operator;
import com.ds.infrastructure.hyperspace.common.config.*;
import com.ds.infrastructure.hyperspace.common.flow.handler.FlowHandler;
import com.ds.infrastructure.hyperspace.container.admin.AdminServerChannelInitializer;
import com.ds.infrastructure.hyperspace.container.admin.ConsoleClientChannelInitializer;
import com.ds.infrastructure.hyperspace.container.endpoint.connector.HyperspaceConnector;
import com.ds.infrastructure.hyperspace.container.endpoint.destination.NettyTargetEndpoint;
import com.ds.infrastructure.hyperspace.container.endpoint.destination.TargetEndpointNode;
import com.ds.infrastructure.hyperspace.container.endpoint.destination.resolver.k8s.K8sTokenStorage;
import com.ds.infrastructure.hyperspace.container.endpoint.entry.NettyEntryEndpoint;
import com.ds.infrastructure.hyperspace.container.flow.FlowPipelineOrchestrator;
import com.ds.infrastructure.hyperspace.container.flow.handler.grpc.GrpcResponseCodePostFlowHandler;
import com.ds.infrastructure.hyperspace.container.flow.handler.http.HttpHeaderSetterPreFlowHandler;
import com.ds.infrastructure.hyperspace.container.flow.handler.loadbalance.RoundRobinWithWeightPreHandler;
import com.ds.infrastructure.hyperspace.container.flow.handler.misc.DynamicDuplexFlowHandler;
import com.ds.infrastructure.hyperspace.container.flow.handler.opentracing.OpentracingMarkSampledPreFlowHandler;
import com.ds.infrastructure.hyperspace.container.flow.handler.route.GrpcServiceRoutePreFlowHandler;
import com.ds.infrastructure.hyperspace.container.flow.handler.route.HttpFullRoutePreFlowHandler;
import com.ds.infrastructure.hyperspace.container.flow.handler.route.HttpRoutePreFlowHandler;
import com.ds.infrastructure.hyperspace.container.flow.handler.security.BlackOrWhiteListPreFlowHandler;
import com.ds.infrastructure.hyperspace.container.flow.handler.security.RateLimiterPreFlowHandler;
import com.ds.infrastructure.hyperspace.container.flow.handler.zone.ConditionalZonePreFlowHandler;
import com.ds.infrastructure.hyperspace.container.flow.handler.zone.CustomHttpExtensionZonePreFlowHandler;
import com.ds.infrastructure.hyperspace.container.flow.handler.zone.DynamicZonePreFlowHandler;
import com.ds.infrastructure.hyperspace.container.flow.handler.zone.StaticZonePreFlowHandler;
import com.ds.infrastructure.hyperspace.container.internal.gzs.GzsService;
import com.ds.infrastructure.hyperspace.container.internal.opentracing.reporter.IOpentracingReporter;
import com.ds.infrastructure.hyperspace.container.internal.opentracing.reporter.OpentracingReporter;
import com.ds.infrastructure.hyperspace.container.internal.operator.*;
import com.ds.infrastructure.hyperspace.container.internal.transport.NettyTransportFactoryDelegate;
import com.ds.infrastructure.hyperspace.extension.DataExtractor;
import com.ds.infrastructure.hyperspace.extension.HttpDataExtractor;
import com.ds.infrastructure.hyperspace.monitor.MetricsManager;
import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoop;
import io.netty.channel.EventLoopGroup;
import io.netty.handler.codec.http.*;
import io.netty.handler.ssl.*;
import io.netty.util.concurrent.EventExecutor;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.io.File;
import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import static com.ds.infrastructure.hyperspace.container.internal.util.StringConstant.COLON;
import static io.netty.channel.ChannelOption.*;

/**
 * @author  Weishaoying
 */
@Slf4j
public class NettyContainer extends AbstractContainer<NettyTargetEndpoint> {

    private static final int MAX_CONNECTION_SIZE = 60000;

    private static final NettyContainer INSTANCE = new NettyContainer();

    public static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

    @Getter
    private GzsService gzsService;

    @Getter
    private NettyTransportFactoryDelegate nettyTransportFactoryDelegate;

    @Getter
    private EventLoopGroup entryAcceptorEventLoopGroup;

    @Getter
    private EventLoopGroup entryWorkerEventLoopGroup;

    @Getter
    private EventLoopGroup targetHealthCheckEventLoopGroup;

    private EventLoopGroup adminBossGroup;

    private EventLoopGroup adminWorkerGroup;

    @Getter
    private EventLoopGroup opentracingReporterEventLoopGroup;

    @Getter
    private EventLoopGroup resolverEventLoopGroup;

    @Getter
    private DnsConfig dnsConfig;

    @Getter
    private K8sConfig k8sConfig;

    @Getter
    private Http2Config http2Config;

    @Getter
    private volatile K8sTokenStorage k8sTokenStorage;

    @Getter
    private IOpentracingReporter opentracingReporter;
    @Getter
    private OpentracingConfig opentracingConfig;

    private AtomicInteger globalEstablishedConnectionSize = new AtomicInteger();

    private int targetMaxGlobalConnectionSize = MAX_CONNECTION_SIZE;

    @Getter
    private volatile ResourceConfig resourceConfig;

    @Getter
    private volatile TargetProtocolConfig protocolConfig;

    @Getter
    private volatile AdminConfig adminConfig;

    @Getter
    @Setter
    private volatile MetricsManager metricsManager;

    @Getter
    private volatile SslContext sslServerContext;

    @Getter
    private FlowPipelineOrchestrator flowPipelineOrchestrator;

    private Map<EventLoop, HyperspaceConnector> hyperspaceConnectorMap = new ConcurrentHashMap<>();

    private Map<String, DataExtractor> httpDataExtractorMap = new ConcurrentHashMap<>();

    private Map<String, Operator> operatorMap = new ConcurrentHashMap<>();

    private Map<String, FlowHandler> flowHandlerMap = new ConcurrentHashMap<>();

    private Map<String, NettyEntryEndpoint> entryEndpointMap = new ConcurrentHashMap<>();

    private Map<String, EndpointProtocolEnum> entryProtocolMap = new ConcurrentHashMap<>();

    private Map<String, NettyTargetEndpoint> targetEndpointMap = new ConcurrentHashMap<>();

    private Map<String, TargetEndpointNode> targetEndpointNodeMap = new ConcurrentHashMap<>();

    private Map<String, AtomicInteger> targetNodeConnectionCounterMap = new ConcurrentHashMap<>();

    private Map<String, AtomicInteger> targetEndpointPendingCounterMap = new ConcurrentHashMap<>();

    private Map<Integer, AtomicInteger> entryEndpointPendingCounterMap = new ConcurrentHashMap<>();

    private NettyContainer() {
    }

    public static NettyContainer getInstance() {
        return INSTANCE;
    }

    @Override
    protected void startInternal(ContainerConfig containerConfig) {
        try {
            OBJECT_MAPPER.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
            this.resourceConfig = containerConfig.getResourceConfig();
            this.protocolConfig = containerConfig.getProtocolConfig();
            this.adminConfig = containerConfig.getAdminConfig();
            this.opentracingConfig = containerConfig.getOpentracingConfig();
            this.dnsConfig = containerConfig.getDnsConfig();
            this.k8sConfig = containerConfig.getK8sConfig();
            this.http2Config = containerConfig.getHttp2Config();
            if (resourceConfig.getTargetMaxGlobalConnectionSize() != null) {
                this.targetMaxGlobalConnectionSize = resourceConfig.getTargetMaxGlobalConnectionSize().intValue();
            }

            initGzsSdk(containerConfig.getGzsConfig());

            HyperspaceThreadFactory adminHyperspaceThreadFactory = new HyperspaceThreadFactory("admin");
            this.nettyTransportFactoryDelegate = new NettyTransportFactoryDelegate();
            this.adminBossGroup = nettyTransportFactoryDelegate.newAcceptorEventLoopGroup(
                    adminHyperspaceThreadFactory, 1);
            this.adminWorkerGroup = nettyTransportFactoryDelegate.newEventLoopGroup(
                    adminHyperspaceThreadFactory, 1);
            this.targetHealthCheckEventLoopGroup = nettyTransportFactoryDelegate.newEventLoopGroup(
                    new HyperspaceThreadFactory("healthcheck"),
                    resourceConfig.getTargetHealthCheckWorkerSize());
            this.opentracingReporterEventLoopGroup = nettyTransportFactoryDelegate.newEventLoopGroup(
                    new HyperspaceThreadFactory("OpentracingSender"), 1);
            this.resolverEventLoopGroup = nettyTransportFactoryDelegate.newEventLoopGroup(
                    new HyperspaceThreadFactory("resolver"), 1);
            HyperspaceThreadFactory proxyHyperspaceThreadFactory = new HyperspaceThreadFactory("proxy");
            this.entryAcceptorEventLoopGroup = nettyTransportFactoryDelegate.newAcceptorEventLoopGroup(
                    proxyHyperspaceThreadFactory, resourceConfig.getAcceptorSize());
            this.entryWorkerEventLoopGroup = nettyTransportFactoryDelegate.newEventLoopGroup(
                    proxyHyperspaceThreadFactory, resourceConfig.getWorkerSize());

            this.k8sTokenStorage = new K8sTokenStorage(k8sConfig);
            this.sslServerContext = initSslServerContext(containerConfig.getTlsConfig());
            this.flowPipelineOrchestrator = new FlowPipelineOrchestrator(this);
            Iterator<EventExecutor> eventExecutorIterator = entryWorkerEventLoopGroup.iterator();
            while (eventExecutorIterator.hasNext()) {
                EventLoop workerEventLoop = (EventLoop) eventExecutorIterator.next();
                hyperspaceConnectorMap.putIfAbsent(workerEventLoop,
                        new HyperspaceConnector(this, workerEventLoop,
                                flowPipelineOrchestrator.getEntryFlowOperation(),
                                flowPipelineOrchestrator.getTargetFlowOperation()));
            }

            initDataExtractor();
            initOperator();
            initFlowHandler();
            initEndpointAndPipeline(containerConfig.getEnv(),
                    containerConfig.getEndpoints(), containerConfig.getPipelines());
            initAdminServer();
            initOpentracing(containerConfig.getOpentracingConfig());

        } catch (Exception e) {
            stopInternal();
            log.error("Init hyperspace container failure!", e);
            throw new ContainerInitException("Init hyperspace failed!", e);
        }
    }

    @Override
    protected void stopInternal() {
        if (targetHealthCheckEventLoopGroup != null) {
            targetHealthCheckEventLoopGroup.shutdownGracefully();
        }

        if (entryAcceptorEventLoopGroup != null) {
            entryAcceptorEventLoopGroup.shutdownGracefully();
        }

        if (entryWorkerEventLoopGroup != null) {
            entryWorkerEventLoopGroup.shutdownGracefully();
        }

        if (adminBossGroup != null) {
            adminBossGroup.shutdownGracefully();
        }

        if (adminWorkerGroup != null) {
            adminWorkerGroup.shutdownGracefully();
        }
        if (resolverEventLoopGroup != null) {
            resolverEventLoopGroup.shutdownGracefully();
        }
        if (opentracingReporterEventLoopGroup != null) {
            opentracingReporterEventLoopGroup.shutdownGracefully();
        }
        if (opentracingReporter != null) {
            opentracingReporter.shutdown();
        }
    }

    @Override
    public NettyTargetEndpoint getTargetEndpoint(String nameSpace) {
        return targetEndpointMap.get(nameSpace);
    }

    @Override
    public DataExtractor getDataExtractor(String id) {
        return httpDataExtractorMap.get(id);
    }

    @Override
    public Operator getOperator(String name) {
        return operatorMap.get(name);
    }

    @Override
    public FlowHandler getFlowHandler(String name) {
        return flowHandlerMap.get(name);
    }

    public List<String> initEndpointAndPipeline(EndpointsConfig endpointsConfig,
                                                List<PipelineConfig> pipelineConfigList) {
        List<String> errorLog = new ArrayList<>();
        try {
            initEntryAndTargetEndpoint(endpointsConfig, errorLog);
            flowPipelineOrchestrator.initPipelines(pipelineConfigList, errorLog);
        } catch (Exception e) {
            log.error("Init endpoint and pipeline failed!", e);
            errorLog.add(e.getMessage());
        }
        return errorLog;
    }

    public boolean isEntryEndpointExists(String nameSpace) {
        return entryEndpointMap.get(nameSpace) != null;
    }

    public EndpointProtocolEnum getEntryEndpointProtocol(String host, Integer port) {
        return entryProtocolMap.get(getEntryNodeKey(host, port));
    }

    public void putEntryEndpointProtocol(String host, Integer port, EndpointProtocolEnum protocolEnum) {
        entryProtocolMap.putIfAbsent(getEntryNodeKey(host, port), protocolEnum);
    }

    public TargetEndpointNode getTargetEndpointNode(String endpointNamespace, TargetNodeConfig targetNodeConfig) {
        return targetEndpointNodeMap.get(TargetEndpointNode.getNodeKey(endpointNamespace, targetNodeConfig));
    }

    public void putTargetEndpointNode(TargetEndpointNode targetEndpointNode) {
        targetEndpointNodeMap.putIfAbsent(targetEndpointNode.getKey(), targetEndpointNode);
        targetNodeConnectionCounterMap.putIfAbsent(targetEndpointNode.getKey(), new AtomicInteger());
    }

    public void removeTargetEndpointNode(TargetEndpointNode targetEndpointNode) {
        targetEndpointNodeMap.remove(targetEndpointNode.getKey());
        targetNodeConnectionCounterMap.remove(targetEndpointNode.getKey());
    }

    public boolean allowIncrementTargetConnection(TargetEndpointNode targetEndpointNode) {
        if (globalEstablishedConnectionSize.incrementAndGet() > targetMaxGlobalConnectionSize) {
            log.warn("Connections exceed max global size {}, target node({})", targetMaxGlobalConnectionSize,
                    targetEndpointNode.getKey());
            globalEstablishedConnectionSize.decrementAndGet();
            return false;
        }

        AtomicInteger nodeConnectionCounter = targetNodeConnectionCounterMap.get(
                targetEndpointNode.getKey());
        int nodeMaxConnectionSize = targetEndpointNode.getNodeConfig()
                .getProtocolConfig().getConnectionMaxPoolSize();
        if (nodeConnectionCounter.incrementAndGet() > nodeMaxConnectionSize) {
            log.warn("Connections exceed max node size {}, target node({})", nodeMaxConnectionSize,
                    targetEndpointNode.getKey());
            globalEstablishedConnectionSize.decrementAndGet();
            nodeConnectionCounter.decrementAndGet();
            return false;
        }
        return true;
    }

    public boolean needHoldTargetConnection(TargetEndpointNode targetEndpointNode) {
        TargetNodeConfig targetNodeConfig = targetEndpointNode.getNodeConfig();
        if (targetNodeConfig.getProtocol().isLongLive() || targetEndpointNode.isShutdown()) {
            return false;
        }
        int nodeInitConnectionSize = targetEndpointNode.getNodeConfig()
                .getProtocolConfig().getConnectionInitialPoolSize();
        AtomicInteger nodeConnectionCounter = targetNodeConnectionCounterMap.get(
                targetEndpointNode.getKey());
        if (nodeConnectionCounter.get() > nodeInitConnectionSize) {
            return false;
        }
        return true;
    }

    public void decrementTargetConnectionSize(TargetEndpointNode targetEndpointNode) {
        globalEstablishedConnectionSize.decrementAndGet();
        AtomicInteger nodeConnectionCounter = targetNodeConnectionCounterMap.get(
                targetEndpointNode.getKey());
        if (nodeConnectionCounter != null) {
            nodeConnectionCounter.decrementAndGet();
        }
    }

    public int getTargetEndpointPendingSize(String targetEndpointNamespace, String zone) {
        AtomicInteger nodePendingSize = targetEndpointPendingCounterMap.get(
                getEndpointPendingKey(targetEndpointNamespace, zone));
        return nodePendingSize == null ? 0 : nodePendingSize.get();
    }

    public void incrementTargetEndpointPendingSize(String targetEndpointNamespace, String zone) {
        AtomicInteger nodePendingSize = targetEndpointPendingCounterMap.computeIfAbsent(
                getEndpointPendingKey(targetEndpointNamespace, zone), key -> new AtomicInteger());
        nodePendingSize.incrementAndGet();
    }

    public void decrementTargetEndpointPendingSize(String targetEndpointNamespace, String zone) {
        AtomicInteger nodePendingSize = targetEndpointPendingCounterMap.get(
                getEndpointPendingKey(targetEndpointNamespace, zone));
        if (nodePendingSize != null && nodePendingSize.get() > 0) {
            nodePendingSize.addAndGet(-1);
        }
    }

    public int getEntryEndpointPendingSize(Integer entryEndpointPort) {
        AtomicInteger entryEndpointPendingSize = entryEndpointPendingCounterMap.
                computeIfAbsent(entryEndpointPort, key -> new AtomicInteger());
        return entryEndpointPendingSize == null ? 0 : entryEndpointPendingSize.get();
    }

    public void incrementEntryEndpointPendingSize(Integer entryEndpointPort) {
        AtomicInteger entryEndpointPendingSize = entryEndpointPendingCounterMap.
                computeIfAbsent(entryEndpointPort, key -> new AtomicInteger());
        entryEndpointPendingSize.incrementAndGet();
    }

    public void decrementEntryEndpointPendingSize(Integer entryEndpointPort) {
        AtomicInteger entryEndpointPendingSize = entryEndpointPendingCounterMap.get(entryEndpointPort);
        if (entryEndpointPendingSize != null && entryEndpointPendingSize.get() > 0) {
            entryEndpointPendingSize.addAndGet(-1);
        }
    }


    public Map<EventLoop, HyperspaceConnector> getHyperspaceConnectorMap() {
        return hyperspaceConnectorMap;
    }

    public HyperspaceConnector getHyperspaceConnector(EventLoop eventLoop) {
        return hyperspaceConnectorMap.get(eventLoop);
    }

    public SslContext initSslServerContext(TlsConfig tlsConfig) {
        if (tlsConfig != null && tlsConfig.getCertificate() != null && tlsConfig.getPrivateKey() != null) {
            try {
                String privateKeyPassword = StringUtils.isEmpty(tlsConfig.getPrivateKeyPassword()) ?
                        null : tlsConfig.getPrivateKeyPassword();
                SslContextBuilder sslContextBuilder = SslContextBuilder.forServer(
                        new File(tlsConfig.getCertificate()),
                        new File(tlsConfig.getPrivateKey()), privateKeyPassword)
                        .sslProvider(SslProvider.OPENSSL)
                        .ciphers(tlsConfig.getCiphers())
                        .applicationProtocolConfig(new ApplicationProtocolConfig(
                                ApplicationProtocolConfig.Protocol.ALPN,
                                ApplicationProtocolConfig.SelectorFailureBehavior.NO_ADVERTISE,
                                ApplicationProtocolConfig.SelectedListenerFailureBehavior.ACCEPT,
                                ApplicationProtocolNames.HTTP_1_1,
                                ApplicationProtocolNames.HTTP_2))
                        .protocols(tlsConfig.getProtocols());
                String trustedCertificate = tlsConfig.getTrustedCertificate();
                if (!StringUtils.isEmpty(trustedCertificate)) {
                    sslContextBuilder.trustManager(new File(trustedCertificate));
                }
                return sslContextBuilder.build();
            } catch (Exception e) {
                log.error("Init ssl context failed!", e);
                throw new ContainerInitException("Endpoint ssl context init failed!", e);
            }
        }
        return null;
    }

    private void initGzsSdk(GzsConfig gzsConfig) {
        if (gzsConfig == null || StringUtils.isEmpty(gzsConfig.getUrl())) {
            log.info("Ignore init gzs sdk, gzs config not found!");
            return;
        }
        try {
            if (CollectionUtils.isEmpty(gzsConfig.getShardingKeyTypeList())) {
                return;
            }
            gzsService = new GzsService(gzsConfig);
            log.info("Init gzs sdk successfully, gzs url [{}]!", gzsConfig.getUrl());
        } catch (Exception excetion) {
            log.error("Init gzs sdk failed. gzs url [{}]", gzsConfig.getUrl(), excetion);
            throw new ContainerInitException("Init gzs sdk failed!", excetion);
        }
    }

    private void initOperator() {
        ServiceLoader<Operator> operatorServiceLoader = ServiceLoader.load(Operator.class);
        for (Operator operator : operatorServiceLoader) {
            operatorMap.putIfAbsent(operator.getOperatorName(), operator);
        }
        EqualOperator equalOperator = new EqualOperator();
        NotEqualOperator notEqualOperator = new NotEqualOperator();
        StartWithOperator startWithOperator = new StartWithOperator();
        EndWithOperator endWithOperator = new EndWithOperator();
        SubStringOperator subStringOperator = new SubStringOperator();
        DelimiterSubStringOperator delimiterSubStringOperator = new DelimiterSubStringOperator();
        DelimiterLastSubStringOperator delimiterLastSubStringOperator = new DelimiterLastSubStringOperator();
        HttpQueryValueOperator httpQueryValueOperator = new HttpQueryValueOperator();
        IndexofOperator indexofOperator = new IndexofOperator();
        StringRemoveOperator stringRemoveOperator = new StringRemoveOperator();
        operatorMap.putIfAbsent(equalOperator.getOperatorName(), equalOperator);
        operatorMap.putIfAbsent(notEqualOperator.getOperatorName(), notEqualOperator);
        operatorMap.putIfAbsent(startWithOperator.getOperatorName(), startWithOperator);
        operatorMap.putIfAbsent(endWithOperator.getOperatorName(), endWithOperator);
        operatorMap.putIfAbsent(subStringOperator.getOperatorName(), subStringOperator);
        operatorMap.putIfAbsent(delimiterSubStringOperator.getOperatorName(), delimiterSubStringOperator);
        operatorMap.putIfAbsent(delimiterLastSubStringOperator.getOperatorName(), delimiterLastSubStringOperator);
        operatorMap.putIfAbsent(httpQueryValueOperator.getOperatorName(), httpQueryValueOperator);
        operatorMap.putIfAbsent(indexofOperator.getOperatorName(), indexofOperator);
        operatorMap.putIfAbsent(stringRemoveOperator.getOperatorName(), stringRemoveOperator);
    }

    private void initDataExtractor() {
        ServiceLoader<HttpDataExtractor> serviceLoader = ServiceLoader.load(HttpDataExtractor.class);
        for (HttpDataExtractor httpDataExtractor : serviceLoader) {
            String id = httpDataExtractor.id();
            if (StringUtils.isEmpty(id) || httpDataExtractorMap.get(id) != null) {
                log.error("HttpDataExtractor id [{}] is null or duplicate", id);
                throw new ContainerInitException("HttpDataExtractor id is null or duplicate");
            }
            httpDataExtractorMap.put(httpDataExtractor.id(), httpDataExtractor);
        }
    }

    private void initFlowHandler() {
        ServiceLoader<FlowHandler> serviceLoader = ServiceLoader.load(FlowHandler.class);
        for (FlowHandler flowHandler : serviceLoader) {
            flowHandlerMap.put(flowHandler.handlerName(), flowHandler);
        }

        StaticZonePreFlowHandler staticZonePreFlowHandler = new StaticZonePreFlowHandler();
        DynamicZonePreFlowHandler dynamicZonePreFlowHandler = new DynamicZonePreFlowHandler();
        CustomHttpExtensionZonePreFlowHandler customHttpExtensionZonePreFlowHandler =
                new CustomHttpExtensionZonePreFlowHandler();
        HttpRoutePreFlowHandler httpRoutePreFlowHandler = new HttpRoutePreFlowHandler();
        HttpFullRoutePreFlowHandler httpFullRoutePreFlowHandler = new HttpFullRoutePreFlowHandler();
        GrpcServiceRoutePreFlowHandler grpcServiceRoutePreFlowHandler = new GrpcServiceRoutePreFlowHandler();
        HttpHeaderSetterPreFlowHandler httpHeaderSetterPreFlowHandler = new HttpHeaderSetterPreFlowHandler();
        RoundRobinWithWeightPreHandler robinWithWeightPreHandler = new RoundRobinWithWeightPreHandler();
        BlackOrWhiteListPreFlowHandler blackOrWhiteListPreFlowHandler = new BlackOrWhiteListPreFlowHandler();
        RateLimiterPreFlowHandler rateLimiterPreFlowHandler = new RateLimiterPreFlowHandler();
        DynamicDuplexFlowHandler dynamicDuplexFlowHandler = new DynamicDuplexFlowHandler();
        OpentracingMarkSampledPreFlowHandler opentracingMarkSampledPreFlowHandler = new OpentracingMarkSampledPreFlowHandler();
        GrpcResponseCodePostFlowHandler grpcResponseCodePostFlowHandler = new GrpcResponseCodePostFlowHandler();
        ConditionalZonePreFlowHandler conditionalZonePreFlowHandler = new ConditionalZonePreFlowHandler();
        flowHandlerMap.put(staticZonePreFlowHandler.handlerName(), staticZonePreFlowHandler);
        flowHandlerMap.put(dynamicZonePreFlowHandler.handlerName(), dynamicZonePreFlowHandler);
        flowHandlerMap.put(customHttpExtensionZonePreFlowHandler.handlerName(), customHttpExtensionZonePreFlowHandler);
        flowHandlerMap.put(httpRoutePreFlowHandler.handlerName(), httpRoutePreFlowHandler);
        flowHandlerMap.put(httpFullRoutePreFlowHandler.handlerName(), httpFullRoutePreFlowHandler);
        flowHandlerMap.put(grpcServiceRoutePreFlowHandler.handlerName(),grpcServiceRoutePreFlowHandler);
        flowHandlerMap.put(httpHeaderSetterPreFlowHandler.handlerName(), httpHeaderSetterPreFlowHandler);
        flowHandlerMap.put(robinWithWeightPreHandler.handlerName(), robinWithWeightPreHandler);
        flowHandlerMap.put(blackOrWhiteListPreFlowHandler.handlerName(), blackOrWhiteListPreFlowHandler);
        flowHandlerMap.put(rateLimiterPreFlowHandler.handlerName(), rateLimiterPreFlowHandler);
        flowHandlerMap.put(dynamicDuplexFlowHandler.handlerName(), dynamicDuplexFlowHandler);
        flowHandlerMap.put(opentracingMarkSampledPreFlowHandler.handlerName(), opentracingMarkSampledPreFlowHandler);
        flowHandlerMap.put(grpcResponseCodePostFlowHandler.handlerName(), grpcResponseCodePostFlowHandler);
        flowHandlerMap.put(conditionalZonePreFlowHandler.handlerName(), conditionalZonePreFlowHandler);
    }

    private void initEndpointAndPipeline(String env, EndpointsConfig endpointsConfig,
                                         List<PipelineConfig> pipelineConfigList) throws Exception {
        if (!adminConfig.isEnablePullConsoleConfig()) {
            List<String> errorLog = initEndpointAndPipeline(endpointsConfig, pipelineConfigList);
            if (!errorLog.isEmpty()) {
                throw new ContainerInitException("Endpoint or pipeline init failed! error message " +
                        errorLog.toString());
            }
            return;
        }
        Bootstrap consoleClientBootstrap = new Bootstrap().group(adminWorkerGroup)
                .channel(nettyTransportFactoryDelegate.socketChannelClass())
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, adminConfig.getConnectConsoleTimeout())
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .handler(new ConsoleClientChannelInitializer(this))
                .remoteAddress(new InetSocketAddress(adminConfig.getConsoleHost(),
                        adminConfig.getConsolePort()));
        ChannelFuture channelFuture = consoleClientBootstrap.connect();
        channelFuture.addListener(future -> {
            if (future.isSuccess()) {
                String consoleConfigUri = new StringBuilder().append(adminConfig.getConsoleReleaseUri())
                        .append(env).toString();
                HttpRequest request = new DefaultFullHttpRequest(
                        HttpVersion.HTTP_1_1, HttpMethod.GET, consoleConfigUri);
                request.headers().set(HttpHeaderNames.HOST, adminConfig.getHost());
                request.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE);
                request.headers().set(HttpHeaderNames.ACCEPT_ENCODING, HttpHeaderValues.GZIP);
                channelFuture.channel().writeAndFlush(request);
            } else {
                log.error("Pull console config failed, can't connect to console!", future.cause());
            }
        });
    }

    private void initEntryAndTargetEndpoint(EndpointsConfig endpointsConfig,
                                            List<String> errorLog) throws Exception {
        if (endpointsConfig == null) {
            log.warn("Init endpoint ignored, config not found!!");
            return;
        }
        List<EntryEndpointConfig> entryEndpoints = endpointsConfig.getEntryEndpoints();
        if (!CollectionUtils.isEmpty(entryEndpoints)) {
            for (EntryEndpointConfig entryEndpointConfig : entryEndpoints) {
                try {
                    String entryEndpointNamespace = entryEndpointConfig.getNamespace();
                    NettyEntryEndpoint entryEndpoint = entryEndpointMap.get(entryEndpointNamespace);
                    if (entryEndpoint == null) {
                        NettyEntryEndpoint newEntryEndpoint = new NettyEntryEndpoint(
                                entryEndpointConfig, this);
                        entryEndpointMap.put(entryEndpointNamespace, newEntryEndpoint);
                    } else {
                        entryEndpoint.reload(entryEndpointConfig);
                    }
                } catch (Exception e) {
                    log.error("Init entry endpoint failed!", e);
                    errorLog.add(e.getMessage());
                }
            }
        } else {
            log.warn("Init entry endpoint config ignored, config not found!");
        }

        List<TargetEndpointConfig> targetEndpoints = endpointsConfig.getTargetEndpoints();
        if (!CollectionUtils.isEmpty(targetEndpoints)) {
            for (TargetEndpointConfig targetEndpointConfig : targetEndpoints) {
                try {
                    String targetEndpointNamespace = targetEndpointConfig.getNamespace();
                    NettyTargetEndpoint targetEndpoint = targetEndpointMap.get(targetEndpointNamespace);
                    if (targetEndpoint == null) {
                        NettyTargetEndpoint newTargetEndpoint = new NettyTargetEndpoint(
                                targetEndpointConfig, this);
                        targetEndpointMap.put(targetEndpointNamespace, newTargetEndpoint);
                    } else {
                        targetEndpoint.reload(targetEndpointConfig);
                        metricsManager.removeTargetNodeStatusGauge(targetEndpointNamespace, Collections.emptySet());
                    }
                } catch (Exception e) {
                    log.error("Init target endpoint failed!", e);
                    errorLog.add(e.getMessage());
                }
            }
        } else {
            log.warn("Init target endpoint config ignored, config not found!");
        }

        Map<String, NettyTargetEndpoint> invalidTargetEndpointMap = new HashMap<>();
        for (Map.Entry<String, NettyTargetEndpoint> targetEndpointEntry : targetEndpointMap.entrySet()) {
            boolean isExistTargetEndpointConfig = false;
            for (TargetEndpointConfig newTargetEndpointConfig : targetEndpoints) {
                if (targetEndpointEntry.getKey().equals(newTargetEndpointConfig.getNamespace())) {
                    isExistTargetEndpointConfig = true;
                    break;
                }
            }
            if (!isExistTargetEndpointConfig) {
                invalidTargetEndpointMap.put(targetEndpointEntry.getKey(), targetEndpointEntry.getValue());
            }
        }
        for (Map.Entry<String, NettyTargetEndpoint> targetEndpointEntry : invalidTargetEndpointMap.entrySet()) {
            NettyTargetEndpoint targetEndpoint = targetEndpointEntry.getValue();
            TargetEndpointConfig invalidTargetEndpointConfig = targetEndpoint.getEndpointConfig();
            targetEndpoint.removeNodeStreamAndManager(invalidTargetEndpointConfig.getNodes(), invalidTargetEndpointConfig);
            targetEndpointMap.remove(targetEndpointEntry.getKey());
            metricsManager.removeTargetNodeStatusGauge(targetEndpointEntry.getKey(), Collections.emptySet());
        }
    }

    private void initAdminServer() {
        if (adminConfig == null) {
            log.error("Init container error, admin server config not found!");
            throw new ContainerInitException("ContainerInitConfiguration not found");
        }
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap()
                    .group(adminBossGroup, adminWorkerGroup)
                    .channel(nettyTransportFactoryDelegate.serverSocketChannelClass())
                    .childOption(SO_KEEPALIVE, true)
                    .childOption(TCP_NODELAY, true)
                    .childOption(ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                    .childHandler(new AdminServerChannelInitializer(this));
            serverBootstrap.bind(adminConfig.getHost(), adminConfig.getPort()).sync();
            log.info("The admin server started successfully, bind to port [{}]", adminConfig.getPort());
        } catch (Exception e) {
            log.error("Start admin server failed! bind to port [{}]", adminConfig.getPort(), e);
            throw new ContainerInitException("Start admin Server failed!", e);
        }
    }

    public void initOpentracing(OpentracingConfig opentracingConfig) {
        metricsManager.opentracingConfigGauge(opentracingConfig.getUrl()
                ,String.valueOf(opentracingConfig.isEnable()),String.valueOf(opentracingConfig.getSampleRate()));
        this.opentracingConfig = opentracingConfig;
        if(this.opentracingReporter==null){
            this.opentracingReporter=new OpentracingReporter(this);
            return;
        }
        this.opentracingReporter.reload(this.opentracingConfig);
    }

    private String getEntryNodeKey(String host, Integer port) {
        return new StringBuilder().append(host).append(COLON).append(port).toString();
    }

    private String getEndpointPendingKey(String endpointNamespace, String zone) {
        return new StringBuilder().append(endpointNamespace).append(COLON).append(zone).toString();
    }

}
